package canal;
import java.net.InetSocketAddress;
import java.util.List;


import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;


public class SimpleCanalClientExample {

    private final static String ip = "192.168.10.168";
    private final static int port = 11111;

    public static void main(String args[]) {
        // 创建链接
        CanalConnector connector = CanalConnectors.
                newSingleConnector(new InetSocketAddress(ip, port), "example", "", "");
        int batchSize = 1000;
        int emptyCount = 0;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            int totalEmptyCount = 120;
            while (emptyCount < totalEmptyCount) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    emptyCount++;
                    System.out.println("empty count : " + emptyCount);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                    }
                } else {
                    emptyCount = 0;
                    // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
                    printEntry(message.getEntries());
                }

                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }

            System.out.println("empty too many times, exit");
        } finally {
            connector.disconnect();
        }
    }

    private static void printEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }

            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }
            //操作类型
            EventType operationType = rowChage.getEventType();
            //数据库名
            String databaseName = entry.getHeader().getSchemaName();
            //表名
            String tableName = entry.getHeader().getTableName();
            
            System.out.println(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    operationType));
            //过滤操作类型，只记录，增，改
            if(operationType.toString().equals("DELETE")) continue;
            //过滤数据库，只记录test数据库
            if(!entry.getHeader().getSchemaName().equals("test")) continue;
            
            //遍历内容
            for (RowData rowData : rowChage.getRowDatasList()) {
                if (operationType == EventType.DELETE) {
                    printColumn(rowData.getBeforeColumnsList());
                } else if (operationType == EventType.INSERT) {
                    //printColumn(rowData.getAfterColumnsList());
                    //printColumn(rowData.getAfterColumnsList());
                    StringBuffer sql = new StringBuffer();
                    sql.append(" insert into meter_user VALUES ( ");
                    List<Column> afterColumnsList = rowData.getAfterColumnsList();
                    for (Column column : afterColumnsList) {
                        String value = column.getValue();
                        if("".equals(value)||value.length()==0){
                            value=null;
                            sql.append( value +",");

                        }else {
                            sql.append("'"+ value +"',");
                        }
                     }
                    sql.deleteCharAt(sql.length()-1);
                    sql.append(")");
                    System.out.println("插入："+sql.toString());

                } else {
                    StringBuffer sql = new StringBuffer();
                    sql.append("update meter_user  set ");
                    String meter_no = null;
                    //System.out.println("-------&gt; before");
                    //printColumn(rowData.getBeforeColumnsList());

                    List<Column> afterColumnsList = rowData.getAfterColumnsList();
                    for (Column column : afterColumnsList) {
                        if(column.getName().equals("meter_no")){
                            meter_no = column.getValue();
                        }
                        //字段有更新
                        if(column.getUpdated()){
                            String s = column.getName()+" = '" +column.getValue() + "' , ";
                            sql.append(s);
                        }
                    }
                    sql.append("  initing_data = null where meter_no = '"+meter_no+"'");
                    System.out.println("更新操作："+sql.toString());
                }
            }
        }
    }

    private static void printColumn(List<Column> columns) {
        for (Column column : columns) {
            //字段有更新
            if(column.getUpdated()){
                System.out.println("字段:"+column.getName()+"   值:"+column.getValue());
            }
         }
    }


}